package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/26
 *
 *
 *
 */
public class Demo13_CEP_Order
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner((o, t) -> o.getEventTime() * 1000);

        KeyedStream<OrderEvent, Long> ds = env.readTextFile("data/OrderLog.csv")
                                             .map(new MapFunction<String, OrderEvent>()
                                             {
                                                 @Override
                                                 public OrderEvent map(String value) throws Exception {
                                                     String[] words = value.split(",");
                                                     return new OrderEvent(
                                                         Long.valueOf(words[0]),
                                                         words[1],
                                                         words[2],
                                                         Long.valueOf(words[3])
                                                     );
                                                 }
                                             }).setParallelism(1)
                                             .assignTimestampsAndWatermarks(watermarkStrategy)
                                              //把同一笔订单的下单和支付放一起
                                             .keyBy(OrderEvent::getOrderId);

        /*
            订单的情况:
                    create ---30min以内--- pay
                    create ---超过30min--- pay  异常
                    create                      正常，不支付
                    pay                        异常，缺少create
                ---------------
                    有create :  以 create 开头
                                    下一条必须在30分钟内到达
                                            继续看下一条是pay :    匹配成功
                                            继续看下一条不是pay :  匹配不成功 (获取不到)

                                    下一条没有在30分钟内到达:
                                            当前create 超时:      在超时数据中获取到


                    无create :  以pay开头

                ------------------
                    希望只处理异常的情况:
                        34741,create,11
                        34741,pay,13

                        匹配到两次:
                                    create---pay
                                    pay

                             如果已经有create-pay ，那么就不要 pay。
                                用skip策略。
         */
        Pattern<OrderEvent, OrderEvent> p1 = Pattern.<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "create".equals(value.getEventType());
                }
            })
            //要求create之后，30分钟内，要到达pay
            .within(Time.minutes(30))
            //可以有create   create  next  pay
            // 没有create    pay
            .optional()
            .next("pay")
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "pay".equals(value.getEventType());
                }
            })
            ;


        SingleOutputStreamOperator<OrderEvent> ds2 = CEP.pattern(ds, p1)
                                                            .process(new MyProcess());

        /*
                使用侧流接收

                OrderEvent(orderId=34768, eventType=create, txId=88snrn932, eventTime=1558430950)  不要

                要
                OrderEvent(orderId=2, eventType=create, txId=abc, eventTime=1558432021)
                OrderEvent(orderId=2, eventType=pay, txId=abc, eventTime=1568432021)
                OrderEvent(orderId=1, eventType=pay, txId=abc, eventTime=1558432021)
         */
        Pattern<OrderEvent, OrderEvent> p2 = Pattern.<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "create".equals(value.getEventType());
                }
            })
            .optional()
            .next("pay")
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "pay".equals(value.getEventType());
                }
            })
            ;

        DataStream<OrderEvent> ds3 = ds2.getSideOutput(new OutputTag<>("ex", TypeInformation.of(OrderEvent.class)));
        ds3.print();

        CEP.pattern(ds3,p2)
           .select(new PatternSelectFunction<OrderEvent, String>()
           {
               /*
                    create.next.pay
                    pay
                */
               @Override
               public String select(Map<String, List<OrderEvent>> match) throws Exception {
                   Long orderId = match.get("pay").get(0).getOrderId();
                   if (!match.keySet().contains("create")){
                       return orderId + "只有pay，缺少create！";
                   }else{
                       return orderId + "超时";
                   }

               }
           })
           .print();

        env.execute();
    }

    private static class MyProcess  extends PatternProcessFunction<OrderEvent, OrderEvent> implements TimedOutPartialMatchHandler<OrderEvent>
    {

        /*
                一条数据调用一次。

            处理正常匹配
                    create ---30min以内--- pay   无需处理
                            {
                                create: OrderEvent(110,create)
                                pay:    OrderEvent(110,pay)
                            }

                    pay                           异常
                           {
                                pay:    OrderEvent(110,pay)
                            }
         */
        @Override
        public void processMatch(Map<String, List<OrderEvent>> match, Context ctx, Collector<OrderEvent> out) throws Exception {
            //pay
            if (!match.keySet().contains("create")) {
                //直接使用侧流输出 只有 pay
                ctx.output(new OutputTag<>("ex",TypeInformation.of(OrderEvent.class)),match.get("pay").get(0));
            }
        }

        /*
            处理超时
                    create ---超过30min--- pay
                    create
         */
        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> match, Context ctx) throws Exception {

                ctx.output(new OutputTag<>("ex",TypeInformation.of(OrderEvent.class)),match.get("create").get(0));

        }
    }
}
